/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package fq.window;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * Skeleton code for the datastream walkthrough
 */
public class Main {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		Obj[] data = new Obj[]{
				new Obj(1L, "a", 1, System.currentTimeMillis()+1000),
				new Obj(1L, "a", 2, System.currentTimeMillis()+2000),
				new Obj(1L, "a", 3, System.currentTimeMillis()+5000),
				new Obj(1L, "a", 4, System.currentTimeMillis()+20000),
				new Obj(1L, "a", 5, System.currentTimeMillis()+30000),
		};

		DataStreamSource<Obj> myData = env.fromElements(data);

		SingleOutputStreamOperator<Obj> a = myData.assignTimestampsAndWatermarks(
				WatermarkStrategy.<Obj>forBoundedOutOfOrderness(Duration.ofSeconds(0)).withTimestampAssigner((event, ts) -> event.timestamp)
		);

		// 按时间混动或者滑动
		/*
		DataStream<SinkObj> alerts = a
				.keyBy(Obj::getName)
//				.window(TumblingEventTimeWindows.of(Time.seconds(4)))
				.window(SlidingEventTimeWindows.of(Time.seconds(2),Time.seconds(1)))
				.process(new FraudDetector())
				.name("fraud-detector");
		*/
		// 按次数混动或者滑动
		DataStream<SinkObj> alerts = a
			.keyBy(Obj::getName)
//				.countWindow(3)
				.countWindow(3,1)
				.process(new CountWindowProcess())
			.name("fraud-detector");

		alerts
			.addSink(new MySink())
			.name("send-alerts");

		env.execute("Fraud Detection");
	}
}
